package com.lvmama.java.rhino.etl.process;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.apache.log4j.Logger;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.alibaba.fastjson.JSON;
import com.lvmama.java.rhino.etl.core.AbstractSparkTemplate;
import com.lvmama.java.rhino.spark.tables.ElasticLogBaseParam;
import com.lvmama.java.rhino.spark.tables.ElasticRequestTimeDetail;
import com.lvmama.java.rhino.spark.tables.LogDetail;
import com.lvmama.java.rhino.spark.tables.RequestSpecialParam;
import com.lvmama.java.rhino.spark.tables.RequestTimeDetail;
import com.lvmama.java.rhino.spark.utils.Constants;
import com.lvmama.java.rhino.spark.utils.ElasticUtils;
import com.lvmama.java.rhino.spark.utils.HdfsUtils;
import com.lvmama.java.rhino.spark.utils.LogProcessorUtil;
import com.lvmama.java.rhino.spark.utils.SaveToHdfsUtils;

/**
 * 手动同步，将日志从HDFS同步到Elastic
 * @author wxliyong
 */
public class SparkSyncDataProcesser extends AbstractSparkTemplate {

	private static final long serialVersionUID = 7499414786777228221L;
	private static final Logger LOGGER = Logger.getLogger(SparkSyncDataProcesser.class);
	private static final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd,HH:mm:ss");
	@Override
	public void excute(String[] args) {
		Map<String, Date> dateInfo = null;
		try {
			dateInfo = parseDate(args);
		} catch (Exception e) {
			LOGGER.error(e.getMessage(), e);
			System.err.println(e.getMessage());
			return;
		}
		Date startTime = dateInfo.get("startTime");
		Date endTime = dateInfo.get("endTime");
		
		String userAccessPathFileUri = SaveToHdfsUtils.getFileUri(SaveToHdfsUtils.TABLE_NAME.USER_ACCESS_PATH, startTime);
		String logDetailFileUri = SaveToHdfsUtils.getFileUri(SaveToHdfsUtils.TABLE_NAME.LOG_DETAIL, startTime);
		String apiRequestTimeDetailFileUri = SaveToHdfsUtils.getFileUri(SaveToHdfsUtils.TABLE_NAME.API_REQUEST_TIME_DETAIL, startTime);
		String requestSpecialParamFileUri = SaveToHdfsUtils.getFileUri(SaveToHdfsUtils.TABLE_NAME.REQUEST_SPECIAL_PARAM, startTime);

		if(!HdfsUtils.getInstance().exists(userAccessPathFileUri)) {
			LOGGER.error("当前日期咱无日志信息");
			System.err.println("当前日期咱无日志信息");
			shutdown(args);
			return;
		}
		DataFrame accessPathDf = sqlContext.read().load(userAccessPathFileUri);
		DataFrame logDetailDf = sqlContext.read().load(logDetailFileUri);
		DataFrame requestTimeDf = sqlContext.read().load(apiRequestTimeDetailFileUri);
		DataFrame specialParamDf = null;
		if(HdfsUtils.getInstance().exists(requestSpecialParamFileUri)) {
			specialParamDf = sqlContext.read().load(requestSpecialParamFileUri);
		}
		String userAccessPathTableName = SaveToHdfsUtils.TABLE_NAME.USER_ACCESS_PATH.name();
		String logDetailTableName = SaveToHdfsUtils.TABLE_NAME.LOG_DETAIL.name();
		String requestTimeTableName = SaveToHdfsUtils.TABLE_NAME.API_REQUEST_TIME_DETAIL.name();
		String specialParamTableName = SaveToHdfsUtils.TABLE_NAME.REQUEST_SPECIAL_PARAM.name();
		logDetailDf.registerTempTable(logDetailTableName);
		requestTimeDf.registerTempTable(requestTimeTableName);
		accessPathDf.registerTempTable(userAccessPathTableName);
		if(specialParamDf != null) {
			specialParamDf.registerTempTable(specialParamTableName);
		}
		String sql = "select * from " + userAccessPathTableName + " where requestTime > " + startTime.getTime() + " and requestTime < " + endTime.getTime();
		DataFrame accessPathResultDf = sqlContext.sql(sql);
		sql = contactSql(LogDetail.class, logDetailTableName, userAccessPathTableName, startTime, endTime);
		DataFrame logDetailResultDf = sqlContext.sql(sql);
		sql = contactSql(RequestTimeDetail.class, requestTimeTableName, userAccessPathTableName, startTime, endTime);
		DataFrame requestTimeResultDf = sqlContext.sql(sql);
		DataFrame specialParamDfResultDf = null;
		if(specialParamDf != null) {
			sql = contactSql(RequestSpecialParam.class, specialParamTableName, userAccessPathTableName, startTime, endTime);
			specialParamDfResultDf = sqlContext.sql(sql);
		}
		List<Row> accessPathRows = accessPathResultDf.collectAsList();
		List<Row> logDetailRows = logDetailResultDf.collectAsList();
		List<Row> requestTimeRows = requestTimeResultDf.collectAsList();
		List<Row> specialParamRows = null;
		if(specialParamDfResultDf != null) {
			specialParamRows = specialParamDfResultDf.collectAsList();
		}
		Map<String, ElasticLogBaseParam> result = new HashMap<String, ElasticLogBaseParam>();
		for (Row row : accessPathRows) {
			String threadName = row.getString(row.fieldIndex("threadName"));
			ElasticLogBaseParam elasticLogInfo = result.get(threadName);
			if(elasticLogInfo == null) {
				elasticLogInfo = new ElasticLogBaseParam();
				result.put(threadName, elasticLogInfo);
			}
			elasticLogInfo.setAppName(row.getString(row.fieldIndex("appName")));
			elasticLogInfo.setLvsessionid(row.getString(row.fieldIndex("lvsessionid")));
			elasticLogInfo.setUserId(row.getString(row.fieldIndex("userId")));
			elasticLogInfo.setMethod(row.getString(row.fieldIndex("method")));
			elasticLogInfo.setVersion(row.getString(row.fieldIndex("version")));
			elasticLogInfo.setUdid(row.getString(row.fieldIndex("udid")));
			elasticLogInfo.setOsVersion(row.getString(row.fieldIndex("osVersion")));
			elasticLogInfo.setLvversion(row.getString(row.fieldIndex("lvversion")));
			elasticLogInfo.setLatitude(row.getString(row.fieldIndex("latitude")));
			elasticLogInfo.setLongitude(row.getString(row.fieldIndex("longitude")));
			elasticLogInfo.setFirstChannel(row.getString(row.fieldIndex("firstChannel")));
			elasticLogInfo.setSecondChannel(row.getString(row.fieldIndex("secondChannel")));
			elasticLogInfo.setIp(row.getString(row.fieldIndex("ip")));
			elasticLogInfo.setRequestTime(row.getLong(row.fieldIndex("requestTime")));
			elasticLogInfo.setResponseTime(row.getLong(row.fieldIndex("responseTime")));
		}
		for(Row row : logDetailRows) {
			String threadName = row.getAs("threadName");
			ElasticLogBaseParam elasticLogInfo = result.get(threadName);
			if(elasticLogInfo == null) {
				elasticLogInfo = new ElasticLogBaseParam();
				result.put(threadName, elasticLogInfo);
			}
			elasticLogInfo.setThreadName(threadName);
			elasticLogInfo.setLogDetail(LogProcessorUtil.splitLog(LogProcessorUtil.gunzip(row.getString(row.fieldIndex("content")))));
		}
		for (Row row : requestTimeRows) {
			String threadName = row.getAs("threadName");
			ElasticLogBaseParam elasticLogInfo = result.get(threadName);
			if(elasticLogInfo == null) {
				elasticLogInfo = new ElasticLogBaseParam();
				result.put(threadName, elasticLogInfo);
			}
			List<ElasticRequestTimeDetail> requestTimeDetail = elasticLogInfo.getRequestTimeDetail();
			if(requestTimeDetail == null) {
				requestTimeDetail = new ArrayList<ElasticRequestTimeDetail>();
				elasticLogInfo.setRequestTimeDetail(requestTimeDetail);
			}
			ElasticRequestTimeDetail requestTime = new ElasticRequestTimeDetail();
			requestTime.setTimeType(row.getString(row.fieldIndex("timeType")));
			requestTime.setRequestName(row.getString(row.fieldIndex("requestName")));
			requestTime.setRequestVersion(row.getString(row.fieldIndex("requestVersion")));
			requestTime.setRequestTime(row.getLong(row.fieldIndex("requestTime")));
			requestTime.setResponseTime(row.getLong(row.fieldIndex("responseTime")));
			requestTime.setMillisecond(row.getLong(row.fieldIndex("millisecond")));
			requestTimeDetail.add(requestTime);
		}
		if(specialParamRows != null) {
			for (Row row : specialParamRows) {
				String threadName = row.getAs("threadName");
				ElasticLogBaseParam elasticLogInfo = result.get(threadName);
				if(elasticLogInfo == null) {
					elasticLogInfo = new ElasticLogBaseParam();
					result.put(threadName, elasticLogInfo);
				}
				elasticLogInfo.setParam1(row.getString(row.fieldIndex("param1")));
				elasticLogInfo.setParam2(row.getString(row.fieldIndex("param2")));
				elasticLogInfo.setParam3(row.getString(row.fieldIndex("param3")));
				elasticLogInfo.setParam4(row.getString(row.fieldIndex("param4")));
				elasticLogInfo.setParam5(row.getString(row.fieldIndex("param5")));
				elasticLogInfo.setParam6(row.getString(row.fieldIndex("param6")));
				elasticLogInfo.setParam7(row.getString(row.fieldIndex("param7")));
				elasticLogInfo.setParam8(row.getString(row.fieldIndex("param8")));
				elasticLogInfo.setParam9(row.getString(row.fieldIndex("param9")));
				elasticLogInfo.setParam10(row.getString(row.fieldIndex("param10")));
			}
		}
		
		Collection<ElasticLogBaseParam> values = result.values();
		for (ElasticLogBaseParam log : values) {
			try {
				ElasticUtils.getInstance().putIndex(Constants.ELASTIC_INDEX_LVTU_INDEX, Constants.ELASTIC_TYPE_BASE_PARAM_TYPE, JSON.toJSONString(log));
				System.out.println("日志存储Elasticsearch完成，ThreadName: " + log.getThreadName());
				LOGGER.info("日志存储Elasticsearch完成，ThreadName: " + log.getThreadName());
			} catch (InterruptedException e) {
				LOGGER.error(e.getMessage(), e);
			} catch (ExecutionException e) {
				LOGGER.error(e.getMessage(), e);
			}
		}
		shutdown(args);
	}
	
	public <T>String[] getFields(Class<T> clazz) {
		Field[] fields = clazz.getDeclaredFields();
		if(fields.length == 0) {
			return null;
		}
		boolean canSerializable = false;
		Class<?>[] dd = clazz.getInterfaces();
		for (Class<?> df : dd) {
			if(df.getName().equals(Serializable.class.getName())) {
				canSerializable = true;
				break;
			}
		}
		if(!canSerializable) {
			throw new RuntimeException("实体必须是可序列化的");
		}
		canSerializable = false;
		for (Field field : fields) {
			if(field.getName().equals("serialVersionUID")) {
				canSerializable = true;
				break;
			}
		}
		int fieldSize = 0;
		if(canSerializable) {
			fieldSize = fields.length-1;
		} else {
			fieldSize = fields.length;
		}
		String[] fieldStr = new String[fieldSize];
		int index = 0;
		for (Field field : fields) {
			if(field.getName().equals("serialVersionUID")) {
				continue;
			}
			fieldStr[index] = field.getName();
			index++;
		}
		return fieldStr;
	}
	
	private <T>String contactSql(Class<T> resultClazz, String driverTable, String joinTable, Date startTime, Date endTime) {
		StringBuffer sb = new StringBuffer();
		sb.append("select ");
		String[] fields = getFields(resultClazz);
		for (int i = 0; i < fields.length; i++) {
			sb.append(driverTable).append(".").append(fields[i]);
			if(i == (fields.length -1)) {
				sb.append(" ");
			} else {
				sb.append(", ");
			}
		}
		sb.append("from ").append(driverTable);
		sb.append(" inner join ").append(joinTable);
		sb.append(" on ").append(driverTable).append(".threadName");
		sb.append(" = ").append(joinTable).append(".threadName where ");
		sb.append(joinTable).append(".requestTime > ").append(startTime.getTime());
		sb.append(" and ").append(joinTable).append(".requestTime < ").append(endTime.getTime());
		return sb.toString();
	}
	
	private Map<String, Date> parseDate(String[] args) throws Exception {
		Map<String, Date> result = new HashMap<String, Date>();
		if(args.length < 3) {
			throw new Exception("缺少起始、结束时间，无法完成同步，系统已停止");
		}
		String startTimeStr = args[1];
		String endTimeStr = args[2];
		Date startTime = null;
		Date endTime = null;
		try {
			startTime = format.parse(startTimeStr);
			endTime = format.parse(endTimeStr);
		} catch (ParseException e) {
			throw new Exception(e);
		}
		if(startTime.after(endTime)) {
			throw new Exception("结束时间必须大于起始时间");
		}
		Calendar startCal = Calendar.getInstance();
		startCal.setTime(startTime);
		Calendar endCal = Calendar.getInstance();
		endCal.setTime(endTime);
		int startDay = startCal.get(Calendar.DAY_OF_YEAR);
		int endDay = endCal.get(Calendar.DAY_OF_YEAR);
		if(startDay != endDay) {
			throw new Exception("当前不支持跨日期操作");
		}
		result.put("startTime", startTime);
		result.put("endTime", endTime);
		return result;
	}
	
	@Override
	public void shutdown(String[] args) {
		if(sc != null) {
			sc.close();
		}
		if(callback != null) {
			callback.call(null);
		}
	}
	
}
